package cn.tecnova.Synchronous

import java.sql.Timestamp
import java.math.BigDecimal


import cn.tecnova.bean.{AppcPublicSentimentEvent, AppcUserSubject, AppmChinaCitys, AppmIndustryType}
import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * description:离线全量同步mysql数据到es
  * PublicSentimentEvent
  * UserSubject
  * ChinaCitys
  * IndustryType
  **/
object OfflineSynchronous {


  Logger.getLogger("org").setLevel(Level.ERROR)
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
//      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._

    val chinaCitysDF = sqLContext.read.jdbc(ConfigHandler.url,"china_citys",ConfigHandler.props)
    val userSubjectDF = sqLContext.read.jdbc(ConfigHandler.url,"user_subject",ConfigHandler.props)
    val publicSentimentEventDF = sqLContext.read.jdbc(ConfigHandler.url,"public_sentiment_event",ConfigHandler.props)
    val industryTypeDF = sqLContext.read.jdbc(ConfigHandler.url,"industry_type",ConfigHandler.props)

    //导入写es相关的包
    import cn.tecnova.bean.IntString._
    import org.elasticsearch.spark.sql._

    val res1 = chinaCitysDF.rdd.map(row => {
      AppmChinaCitys(
        row.getAs[String]("id"),
        row.getAs[BigDecimal]("CITY_ID").toString(),
        row.getAs[BigDecimal]("SUPERIOR_ID").toString(),
        row.getAs[String]("CITY_NAME"),
        row.getAs[String]("TYPE"),
        row.getAs[String]("COUNTRY"),
        row.getAs[Int]("LEVEL").toStringPlusInt,
        row.getAs[Int]("ISLAST").toStringPlusInt,
        row.getAs[String]("FOREIGN_NAME"))
    }).toDF()
    res1.saveToEs("appm_china_citys" + "/appm_china_citys")



    val res2 = userSubjectDF.map(row=>{
      AppcUserSubject(
        row.getAs[String]("id"),
        row.getAs[String]("user_id"),
        row.getAs[Int]("subject_type").toStringPlusInt,
        row.getAs[String]("full_name"),
        row.getAs[String]("short_name"),
        row.getAs[String]("modify_name"),
        row.getAs[String]("rel_subject_id"),
        row.getAs[String]("rel_subject_name"),
        row.getAs[String]("response_person"),
        row.getAs[String]("industry"),
        row.getAs[String]("egistered_address"),
        row.getAs[String]("industry_fund"),
        row.getAs[Int]("focus_flag").toStringPlusInt,
        row.getAs[String]("focus_reason"),
        row.getAs[Int]("weight").toStringPlusInt,
        row.getAs[String]("description"),
        row.getAs[String]("microblog"),
        row.getAs[String]("wechat"),
        row.getAs[String]("qq"),
        row.getAs[String]("main_source"),
        row.getAs[String]("created_time"),
        row.getAs[String]("update_user_id"),
        row.getAs[String]("updated_time"))
    })
      res2.saveToEs("appc_user_subject" + "/appc_user_subject")

    val res3 = industryTypeDF.map(row=>{
      AppmIndustryType(
        row.getAs[String]("id"),
        row.getAs[String]("father_code"),
        row.getAs[String]("code"),
        row.getAs[String]("name"),
        row.getAs[String]("description"))
    })
    res3.saveToEs("appm_industry_type" + "/appm_industry_type")

    val res4 = publicSentimentEventDF.map(row=>{
      AppcPublicSentimentEvent(
        row.getAs[String]("id"),
        row.getAs[String]("user_id"),
        row.getAs[String]("article_id"),
        row.getAs[String]("article_title"),
        row.getAs[Timestamp]("created_time").toString,
        row.getAs[Timestamp]("end_time").toString,
        row.getAs[String]("description"),
        row.getAs[String]("keywords"),
        row.getAs[Int]("run_flag").toStringPlusInt,
        row.getAs[Timestamp]("created_time").toString,
        row.getAs[String]("update_user_id"),
        row.getAs[Timestamp]("updated_time").toString,
        row.getAs[Int]("del_flag").toStringPlusInt
      )
    })
    res4.saveToEs("appc_public_sentiment_event" + "/appc_public_sentiment_event")

    sc.stop()

  }

}
